package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.util.Date;

/*
从Kafka中读取ods层用户行为数据
 */
public class BaseLogApp {
    //定义用户行为主题信息
    private static final String TOPIC_START="dwd_start_log";
    private static final String TOPIC_PAGE="dwd_page_log";
    private static final String TOPIC_DISPLAY="dwd_display_log";

    public static void main(String[] args)  throws  Exception{
        // TODO 0. 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度，这里和Kafka分区数目保持一致
        env.setParallelism(3);
        //设置CK参数： 设置精准一次性保证(默认)
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        //checkpoint 必须在一分钟内完成，否则就会被抛弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop:8020/gmall/flink/checkpoint"));
//        System.setProperty("HADOOP_USER_NAME","root");

        //指定消费者配置信息
        String groupId = "ods_dwd_base_log_app";
        String topic ="ods_base_log";

        //TODO 1. 从kafka中读取数据
        FlinkKafkaConsumer<String> kafakSource = MyKafkaUtil.getKafkaSource(topic,groupId);
        DataStreamSource<String> kafkaDS = env.addSource(kafakSource);
        // 转换为Json对象
        SingleOutputStreamOperator<JSONObject> jsonObjectDS = kafkaDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(s);
                        return jsonObject;
                    }
                }
        );
        //打印测试
//        jsonObjectDS.print();

        //TODO 2. 识别新老访客
        // 在日志中有is_new参数 is_new=1是新用户，但是如果被重置会导致is_new 重置
        //按照mid进行分组，保存mid某天访问情况(将首次访问日期作为状态保存起来)，等后面该设备
        //再有日志过来的时候，从状态中获取日期和日志产生时间进行对比
        KeyedStream<JSONObject,String> midKeyedDS = jsonObjectDS.keyBy(
                data -> data.getJSONObject("common").getString("mid")
        );

        //检验采集到的数据是新老访客
        SingleOutputStreamOperator<JSONObject> midWithNewFlagDS = midKeyedDS.map(
                new RichMapFunction<JSONObject, JSONObject>() {
                    //声明第一次访问日期的状态
                    private ValueState<String> firstVisitDataState;
                    //声明日期数据格式化对象
                    private SimpleDateFormat simpleDateFormat;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //初始化数据
                        firstVisitDataState = getRuntimeContext().getState(
                                new ValueStateDescriptor<String>("newMidDateState",String.class)
                        );
                        simpleDateFormat = new SimpleDateFormat("yyyyMMdd");
                    }

                    @Override
                    public JSONObject map(JSONObject jsonObject) throws Exception {
                        //打印数据
//                        System.out.println(jsonObject);
//                        // 获取访问路径，0表示老访客，1表示新访问客
//                        String isNew = jsonObject.getJSONObject("common").getString("is_new");
//                        Long ts = jsonObject.getLong("ts");
//
//                        //判断标记如果为1 ，则继续校验数据
//                        if ("1".equals(isNew)){
//                            //获取新访客状态
//                            String newMidDate = firstVisitDataState.value();
//                            // 获取当前数据访问日期
//                            String  tsDate = simpleDateFormat.format(new Date(ts));
//
////                            System.out.println(newMidDate+"-->"+tsDate);
//                            //如果新访客状态不为空，说明该设备已经访问过，则将范文标记设置为0
//                            if (newMidDate !=null &&newMidDate.length() !=0) {
//                                isNew = "0";
//                                jsonObject.getJSONObject("common").put("is_new", isNew);
//                                firstVisitDataState.update(newMidDate);
//                            }else{
//                                //如果复检后，该设备的确没有访问过，那么更新状态为当前日志
//                                firstVisitDataState.update(tsDate);
//                            }
//                            //TODO: 也有可能在is_new=0的情况下是错误的
//                        }else{
//                            //获取旧访客状态
//                            String oldMidDate = firstVisitDataState.value();
//                            // 获取当前数据访问日期
//                            String  tsDate = simpleDateFormat.format(new Date(ts));
//
////                            System.out.println(oldMidDate+"-->"+tsDate);
//                            //如果旧访客状态不为空，说明该设备已经访问过，确实是旧访客
//                            if (oldMidDate !=null &&oldMidDate.length() !=0){
//                                firstVisitDataState.update(tsDate);
//
//                            }else{
//                                isNew ="1";
//                                jsonObject.getJSONObject("common").put("is_new",isNew);
//                                firstVisitDataState.update(tsDate);
//
//                            }
//                        }

                    //返回确认过新老访客的json数据
                        return jsonObject;
                    }
                }
        );
//        midWithNewFlagDS.print();


        //TODO 3. 利用侧输出流实现数据拆分
        // 定义启动和曝光数据的侧输出流标签，
        // 侧输出流两个作用： 分流，处理延迟数据
        OutputTag<String> startTag = new OutputTag<String>("start"){};
        OutputTag<String> displayTag = new OutputTag<String>("display"){};

        //页面访问日志，启动日志，曝光日志
        SingleOutputStreamOperator<String> pageDStream = midWithNewFlagDS.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<String> collector) throws Exception {
                        // 获取数据中的启动相关字段
                        JSONObject startJsonObject = jsonObject.getJSONObject("start");
                        // 将数据转换为字符串，准备向流中输出
                        String dataStr = jsonObject.toString();

                        //如果是启动日志，输出到启动侧输出流
                        if (startJsonObject != null && startJsonObject.size() > 0) {
                            context.output(startTag, dataStr);
                        } else {
                            // TODO： BUG修复，曝光日志也输出页面日志的一种
                            // 如果不是启动日志，就是页面日志，就输出到主流
                            collector.collect(dataStr);

                            // 获取数据中的曝光数据，如果不为空，则将每条曝光日志取出输出到曝光日志侧输出流
                            JSONArray displays = jsonObject.getJSONArray("displays");

                            if (displays != null && displays.size() > 0) {
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject displayJsonObj = displays.getJSONObject(i);
                                    //获取页面id
                                    String pageId = jsonObject.getJSONObject("page").getString("page_id");
                                    displayJsonObj.put("page_id", pageId);
                                    // 将曝光数据输出到侧输出流
                                    context.output(displayTag, displayJsonObj.toString());
                                }
                            }
                        }
                    }
                }
        );


        //获取侧输出流
        DataStream<String> startDStream = pageDStream.getSideOutput(startTag);
        DataStream<String> displayDStream = pageDStream.getSideOutput(displayTag);

        //打印测试
        pageDStream.print("page");
        startDStream.print("start");
        displayDStream.print("display");

        // TODO 4. 将输出输出到kafka不同的主题中
        FlinkKafkaProducer<String> startSink= MyKafkaUtil.getKafkaSink(TOPIC_START);
        FlinkKafkaProducer<String> pageSink = MyKafkaUtil.getKafkaSink(TOPIC_PAGE);
        FlinkKafkaProducer<String> displaySink = MyKafkaUtil.getKafkaSink(TOPIC_DISPLAY);


        startDStream.addSink(startSink);
        pageDStream.addSink(pageSink);
        displayDStream.addSink(displaySink);

        env.execute("dwd_base_log Job");

    }
}
